"""
Case Type   : 基础功能
Case Name   : 创建发布订阅并对发布表进行闪回操作（发布订阅均存在闪回表）
Create At   : 2022/2/7
Owner       : opentestcase026
Description :
    1.设置闪回参数
    2.两个集群创建表
    3.创建函数获取csn及timecapsule
    4.创建发布订阅
    5.更新数据,并查询更新
    6.闪回查询
    7.闪回表
    8.订阅端闪回表
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.插入数据成功,订阅端按顺序进行更新成功
    6.集群A查询结果为空,集群B查询结果不变为1,99,3
    7.t1查询内容为空,第二次闪回后查询结果为1,2,3
    8.集群B更新数据成功,且闪回后数据内容为1,3,99.集群Aupdate成功,集群B数据同步
    为100,3,99
History     :
    Modified by opentestcase012 2022/8/4:增加基础数据同步功能后订阅端有基础数据的同时
    再在订阅端修改表数据,主键冲突,关闭基础数据同步
    Modified by opentestcase004 2022/8/9:开启参数enable_recyclebin后，清理环境需添加
    清理回收站步骤
    Modified by opentestcase012 2022/10/12:优化用例和断言,添加查看数据库状态
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.tb_name1 = 'tb_pubsub_074_1'
        self.tb_name2 = 'tb_pubsub_074_2'
        self.subname1 = "sub_074_1"
        self.pubname1 = "pub_074_1"
        self.hba_pub = os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')
        self.hba_sub = os.path.join(macro.DB_INSTANCE_PATH_REMOTE1,
                                    'pg_hba.conf')
        self.hba_bak_pub = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH), 'pg_hba.conf')
        self.hba_bak_sub = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1), 'pg_hba.conf')
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'

        self.undo_retention_time = self.com_pub.show_param(
            "undo_retention_time")
        self.recyclebin_retention_time = self.com_pub.show_param(
            "recyclebin_retention_time")
        self.enable_recyclebin = self.com_pub.show_param("enable_recyclebin")
        self.enable_default_ustore_table = self.com_pub.show_param(
            "enable_default_ustore_table")

        self.undo_retention_time_sub = self.com_sub.show_param(
            "undo_retention_time", macro.DB_ENV_PATH_REMOTE1)
        self.recyclebin_retention_time_sub = self.com_sub.show_param(
            "recyclebin_retention_time", macro.DB_ENV_PATH_REMOTE1)
        self.enable_recyclebin_sub = self.com_sub.show_param(
            "enable_recyclebin", macro.DB_ENV_PATH_REMOTE1)
        self.enable_default_ustore_table_sub = self.com_sub.show_param(
            "enable_default_ustore_table", macro.DB_ENV_PATH_REMOTE1)
        text = '-----预置条件1:备份配置文件pg_hba.conf expect:状态正常-----'
        self.log.info(text)
        cmd = f'cp {self.hba_pub} {self.hba_bak_pub};'
        self.log.info(cmd)
        msg1 = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        cmd = f'cp {self.hba_sub} {self.hba_bak_sub};'
        self.log.info(cmd)
        msg2 = self.com_pub.get_sh_result(self.pri_userdb_sub, cmd)
        self.assertEqual('', msg1 and msg2, '执行失败:' + text)

        text = '-----预置条件2:查看发布端和订阅端的数据库状态 expect:状态正常-----'
        self.log.info(text)
        status = self.commsh_pub.get_db_cluster_status()
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        text = '-----预置条件3:修改发布端pg_hba文件和配置参数wal_level expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)

        text = '-----预置条件4:订阅端修改pg_hba文件 expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '--step1:设置闪回参数 expect:成功--'
        self.log.info(text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'undo_retention_time=3h')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'recyclebin_retention_time=3h')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'enable_recyclebin=on')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'enable_default_ustore_table=on')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(False)
        self.assertTrue(result, '执行失败:' + text)
        status = self.commsh_pub.get_db_cluster_status()
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status,
                      '执行失败:' + text)

        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'undo_retention_time=3h',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'recyclebin_retention_time=10000',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'enable_recyclebin=on',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'enable_default_ustore_table=on',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.restart_db_cluster(
            False, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status,
                      '执行失败:' + text)

        text = '--step2:两个集群创建表 expect:成功--'
        self.log.info(text)
        create_sql = f"drop table if exists {self.tb_name1};" \
            f"create table {self.tb_name1}(a int primary key) ;" \
            f"drop table if exists {self.tb_name2};" \
            f"create table {self.tb_name2}(id int  primary key, snaptime " \
            f"timestamptz, snapcsn bigint);"
        result = self.commsh_pub.execut_db_sql(
            create_sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         4, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(create_sql,
                                               self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         4, '执行失败:' + text)

        text = '--step3:创建函数获取csn及timecapsule expect:成功--'
        self.log.info(text)
        sql = f"CREATE OR REPLACE FUNCTION findCsn(int8)  " \
            f"RETURNS INTEGER  LANGUAGE plpgsql " \
            f"AS \$BODY\$ declare  count integer;" \
            f"begin" \
            f"  count = (select snapcsn from {self.tb_name2} " \
            f"where id =  \$1);  return count;end;\$BODY\$;" \
            f"CREATE OR REPLACE FUNCTION findTime(int8)  " \
            f"RETURNS timestamptz" \
            f"  LANGUAGE plpgsql AS \$BODY\$ declare  " \
            f"count timestamptz;begin" \
            f"  count = (select snaptime from " \
            f"{self.tb_name2} where id=\$1);  " \
            f"return count;end;\$BODY\$;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(
            2, result.count(self.constant.CREATE_FUNCTION_SUCCESS_MSG),
            '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql,
                                               self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = "--step4:创建发布订阅 expect:成功--"
        self.log.info(text)
        sql = f"drop publication if exists {self.pubname1};" \
            f"create publication {self.pubname1} for all tables ;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"drop subscription if exists {self.subname1};" \
            f"create subscription {self.subname1} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"publication {self.pubname1} with (copy_data=false);"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.create_sub_succ_msg), 1,
                         '执行失败:' + text)

        text = "--step5:更新数据,并查询更新 expect:插入数据成功,订阅端按顺序进" \
               "行更新成功--"
        self.log.info(text)
        sql = f"select pg_sleep(10);insert into {self.tb_name2} " \
            f"select 1, now(), int8in(xidout(next_csn)) " \
            f"from gs_get_next_xid_csn();" \
            f"insert into {self.tb_name1} values(1),(2),(3);" \
            f"select pg_sleep(10);" \
            f"insert into {self.tb_name2} select 2, now(), " \
            f"int8in(xidout(next_csn)) from gs_get_next_xid_csn();" \
            f"update {self.tb_name1} set a = 99 where a = 2;" \
            f"select pg_sleep(10);" \
            f"insert into {self.tb_name2} select 3, " \
            f"now(), int8in(xidout(next_csn)) from gs_get_next_xid_csn();"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.INSERT_SUCCESS_MSG),
                         4, '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1],
                         result, '执行失败:' + text)
        sql_select = f"select * from {self.tb_name1} order by a;" \
            f"select * from {self.tb_name2};"
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('3 rows'), 2, '执行失败:' + text)
        self.assertIn('99', result, '执行失败:' + text)
        self.assertIn('1', result, '执行失败:' + text)
        self.assertIn('3', result, '执行失败:' + text)
        result_a = self.commsh_pub.execut_db_sql(sql_select,
                                                 sql_type=self.user_param_pub)
        self.log.info(result_a)
        self.assertEqual(result, result_a, '执行失败:' + text)

        text = "--step6:闪回查询 expect:集群A查询结果为空,集群B查询结果不变为" \
               "1,99,3--"
        self.log.info(text)
        sql = f"select * from {self.tb_name1} timecapsule " \
            f"csn findCsn(1) order by a;" \
            f"select * from {self.tb_name1} timecapsule " \
            f"timestamp findTime(1) order by a;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn('0 rows', result, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('3 rows', result, '执行失败:' + text)
        self.log.info(result)

        text = "--step7:闪回表 expect:t1查询内容为空,第二次闪回后查询结果为" \
               "1,2,3--"
        self.log.info(text)
        sql = f"timecapsule table {self.tb_name1} to csn findCsn(1);"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.timecapsule_table_succ_msg,
                      result, '执行失败:' + text)
        result_a = self.commsh_pub.execut_db_sql(sql_select,
                                                 sql_type=self.user_param_pub)
        self.log.info(result_a)
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('0 rows', result, '执行失败:' + text)
        self.assertEqual(result, result_a, '执行失败:' + text)
        sql = f"timecapsule table {self.tb_name1} to timestamp findTime(2);"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.timecapsule_table_succ_msg,
                      result, '执行失败:' + text)
        result_a = self.commsh_pub.execut_db_sql(sql_select,
                                                 sql_type=self.user_param_pub)
        self.log.info(result_a)
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('3 rows', result, '执行失败:' + text)
        self.assertEqual(13, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('1', result.splitlines()[2], '执行失败:' + text)
        self.assertIn('2', result.splitlines()[3], '执行失败:' + text)
        self.assertIn('3', result.splitlines()[4], '执行失败:' + text)
        self.assertEqual(result, result_a, '执行失败:' + text)

        text = "--step8:订阅端闪回表 expect:集群B更新数据成功,且闪回后数据内容" \
               "为1,3,99." \
               "集群Aupdate成功,集群B数据同步为100,3,99--"
        self.log.info(text)
        sql = f"update {self.tb_name1} set a = 99 where a = 2;" \
            f"insert into {self.tb_name2} select 4, now(), " \
            f"int8in(xidout(next_csn)) from gs_get_next_xid_csn();" \
            f"insert into {self.tb_name1} values(4);" \
            f"timecapsule table {self.tb_name1} to csn findCsn(4);"
        result = self.commsh_sub.execut_db_sql(sql,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('3 rows', result, '执行失败:' + text)
        self.assertEqual(14, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('1', result.splitlines()[2], '执行失败:' + text)
        self.assertIn('3', result.splitlines()[3], '执行失败:' + text)
        self.assertIn('99', result.splitlines()[4], '执行失败:' + text)
        sql = f"update {self.tb_name1} set a=100 where a=1;"
        result_a = self.commsh_pub.execut_db_sql(sql,
                                                 sql_type=self.user_param_pub)
        self.log.info(result_a)
        result_a = self.commsh_pub.execut_db_sql(sql_select,
                                                 sql_type=self.user_param_pub)
        self.log.info(result_a)
        self.assertEqual(14, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('2', result_a.splitlines()[2], '执行失败:' + text)
        self.assertIn('3', result_a.splitlines()[3], '执行失败:' + text)
        self.assertIn('100', result_a.splitlines()[4], '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(14, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('3', result.splitlines()[2], '执行失败:' + text)
        self.assertIn('99', result.splitlines()[3], '执行失败:' + text)
        self.assertIn('100', result.splitlines()[4], '执行失败:' + text)

    def tearDown(self):
        self.log.info('-----清理环境-----')
        text1 = '-----删除发布订阅 expect:成功-----'
        self.log.info(text1)
        sql = f"drop subscription if exists {self.subname1};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)
        sql = f"drop publication if exists {self.pubname1};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text2 = '-----订阅端和发布端删除表 expect:成功-----'
        self.log.info(text2)
        sql = f"drop table if exists {self.tb_name2};" \
            f"drop table if exists {self.tb_name1};" \
            f"drop function if exists findCsn;" \
            f"drop function if exists findTime;" \
            f"purge recyclebin;"
        sql_drop_sub = self.commsh_sub.execut_db_sql(sql, self.user_param_sub,
                                                     None,
                                                     macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sql_drop_sub)
        sql_drop_pub = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(sql_drop_pub)

        text3 = '-----订阅端和发布端恢复pg_hba.conf文件 expect:成功-----'
        self.log.info(text3)
        cmd = f'if [[ -e {self.hba_pub} && -e {self.hba_bak_pub} ]]; ' \
            f'then rm -rf {self.hba_pub}; ' \
            f'mv {self.hba_bak_pub} {self.hba_pub}; ' \
            f'else echo not exits; fi'
        hba_msg_pub = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        cmd = f'if [[ -e {self.hba_sub} && -e {self.hba_bak_sub} ]]; ' \
            f'then rm -rf {self.hba_sub}; ' \
            f'mv {self.hba_bak_sub} {self.hba_sub}; ' \
            f'else echo not exits; fi'
        hba_msg_sub = self.com_pub.get_sh_result(self.pri_userdb_sub, cmd)

        text4 = '-----订阅端和发布端恢复参数配置 expect:成功-----'
        self.log.info(text4)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        result1 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'undo_retention_time={self.undo_retention_time}')
        result2 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'recyclebin_retention_time={self.recyclebin_retention_time}')
        result3 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_recyclebin={self.enable_recyclebin}')
        result4 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_default_ustore_table={self.enable_default_ustore_table}')
        result5 = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'undo_retention_time={self.undo_retention_time_sub}',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        result6 = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'recyclebin_retention_time={self.recyclebin_retention_time_sub}',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        result7 = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_recyclebin={self.enable_recyclebin_sub}',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        result8 = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_default_ustore_table='
            f'{self.enable_default_ustore_table_sub}',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)

        text5 = '-----重启数据库查看数据库状态 expect:成功-----'
        self.log.info(text5)
        res_msg_pub = self.commsh_pub.restart_db_cluster(False)
        status_pub = self.commsh_pub.get_db_cluster_status()
        self.log.info(status_pub)
        res_msg_sub = self.commsh_sub.restart_db_cluster(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        status_sub = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status_sub)

        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text1)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text1)
        self.assertEqual(2, sql_drop_sub.count(
            self.constant.TABLE_DROP_SUCCESS), '执行失败' + text2)
        self.assertEqual(2, sql_drop_pub.count(
            self.constant.TABLE_DROP_SUCCESS), '执行失败' + text2)
        self.assertEqual(2, sql_drop_sub.count(
            self.constant.DROP_FUNCTION_SUCCESS_MSG), '执行失败' + text2)
        self.assertEqual(2, sql_drop_pub.count(
            self.constant.DROP_FUNCTION_SUCCESS_MSG), '执行失败' + text2)
        self.assertIn('PURGE RECYCLEBIN', sql_drop_sub, '执行失败' + text2)
        self.assertIn('PURGE RECYCLEBIN', sql_drop_pub, '执行失败' + text2)
        self.assertEqual('', hba_msg_pub and hba_msg_sub, '执行失败:' + text3)
        self.assertTrue(result and result1 and result2 and
                        result3 and result4 and result5 and result6
                        and result7 and result8, '执行失败:' + text4)
        self.assertTrue(res_msg_pub and res_msg_sub, '执行失败:' + text5)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status_pub,
                      '执行失败:' + text5)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status_sub,
                      '执行失败:' + text5)
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
